package com.ibm.spatiotemp.server.messaging.pubsub;

import org.apache.log4j.Logger;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.exceptions.JedisConnectionException;
import redis.clients.jedis.exceptions.JedisDataException;

/**
 * @author Yaoliang Chen
 * @version May 20, 2014 Redis based pubsub
 */

public class RedisPubSub implements IPubSub {
	/**
	 * Reference for log4j
	 */
	private static Logger logger = Logger.getLogger(RedisPubSub.class);
	Jedis jedisClient;
	JedisPubSubHandler handler;
	Thread t;
	/*
	 * (non-Javadoc)
	 * 
	 * @see com.WhaleDB.tools.redis.IPubSub#init(com.WhaleDB.tools.LocalConfig,
	 * int)
	 */
	public boolean init(String host, String port) {
		try {
			jedisClient = new Jedis(host, Integer.parseInt(port), 0);
			/* interested in expired keys */
			//jedisClient.configSet("notify-keyspace-events", "Ex");
			jedisClient.connect();
		} catch (JedisConnectionException e) {
			return false;
		}
		handler = new JedisPubSubHandler();
		return true;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.WhaleDB.tools.redis.IPubSub#subscribe(java.lang.String)
	 */
	public boolean psubscribe(String pattern) {
		try {
			t= new Thread(new JedisSubThread(2, jedisClient, handler,
					pattern));
			t.start();
			Thread.sleep(500);
			if (t.isAlive())
				return true;
		} catch (JedisConnectionException e) {
			logger.error("error psubscribe " + pattern + ".\n" + e.getMessage());
		} catch (InterruptedException e) {
			logger.error("error psubscribe " + pattern + ".\n" + e.getMessage());
		}
		return false;
	}

	public boolean subscribe(String... topic) {
		try {
			Thread t = new Thread(new JedisSubThread(2, jedisClient, handler,
					topic));
			t.start();
			Thread.sleep(500);
			if (t.isAlive())
				return true;
		} catch (JedisConnectionException e) {
			logger.error("error subscribe " + topic + ".\n" + e.getMessage());
		} catch (InterruptedException e) {
			logger.error("error subscribe " + topic + ".\n" + e.getMessage());
		}
		return false;
	}

	public String[] getMessage() {
		return handler.getMessage();
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.WhaleDB.tools.redis.IPubSub#publish(java.lang.String,
	 * java.lang.String)
	 */
	public boolean publish(String topic, String values) {
		int retry = 0;
		String lastErr = null;
		while (retry < 10) {
			try {
				jedisClient.publish(topic, values);
				return true;
			} catch (JedisConnectionException e) {
				try {
					Thread.sleep(100);
				} catch (InterruptedException e1) {
				}
				lastErr = e.getMessage();
				logger.warn("retry to publish at " + topic + ". "
						+ e.getClass().toString());
			} catch (JedisDataException de) {
				try {
					Thread.sleep(100);
				} catch (InterruptedException e1) {
				}
				lastErr = de.getMessage();
				logger.fatal("retry to publish at " + topic + ". "
						+ de.getClass().toString() + "," + de.getMessage());
				/* reconnect? */
			}
			retry++;
		}
		logger.error("error publish " + topic + ".\n" + lastErr);
		return false;
	}

	@Override
	public void close() {
		if (t.isAlive())
		{
			t.stop();
			t = null;
		}
		jedisClient.close();
	}
}
